package com.neusoft.dibuy.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

public class ConsumerApp {
    public static void main(String[] args) {
        /**
         * TODO :  s生产者
         */

        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(config);
        /**
         * TODO : 定义主题
         *
         *
         */
        consumer.subscribe(java.util.Arrays.asList("test"));
        /**
         * TODO : 拉取数据
         */
        while (true) {
            consumer.poll(Duration.ofMillis(10))
                    .forEach(record -> {
                        System.out.println(record.key() + ":" + record.value());
                    });
        }

        /**
         * TODO : 关闭资源
         */


    }
}
